/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl.kryo;

import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;
import org.sabayframework.cluster.Cluster;
import org.sabayframework.cluster.ClusterContext;
import org.sabayframework.cluster.ComputerId;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.cluster.NodeId;
import org.sabayframework.log.Log;

/**
 * An activity to be executed on a remote server.
 * TODO support multiple servers.
 * @author Alex Antonau
 * */
public class DistributedActivityContext implements ActivityMonitor<Object, Object>{

	private static AtomicInteger nextId = new AtomicInteger();
	private final int activityId;
	private Set<ComputerId> serverIds;
	private ComputerId clientId;
	private DistributedActivity<Object, Object> distributedActivity;
	private Channel<Object> inputChannel;
	private Channel<Object> outputChannel;
	
	private volatile boolean alive = true;
	private final long startTime;
	private long stopTime;
	private long executionTime;
	private long callCount;
	private long lastRequestStartTime;
	
	
	public DistributedActivityContext(int activityId) {
		super();
		this.activityId = activityId;
		this.startTime = System.currentTimeMillis();
		this.serverIds = Collections.synchronizedSet(new HashSet<ComputerId>());
	}
	public DistributedActivityContext() {
		this(nextId.incrementAndGet());
	}
	
	public int getActivityId() {
		return activityId;
	}
	
	public ComputerId selectRemoteNode(Object data){
		Cluster cluster=ClusterContext.getClusteringService().getCluster();
		NodeId serverId = distributedActivity.getDistributionStrategy().select(cluster, distributedActivity, data);
		return (ComputerId) serverId;
	}
	/**
	 * Returns true if this activity was executed on a given server.
	 * */
	public boolean hasServerId(ComputerId serverId){
		return serverIds.contains(serverId);
	}
	public ComputerId[] getServerIds() {
		return serverIds.toArray(new ComputerId[0]);
	}
	public void addServerId(ComputerId serverId) {
		this.serverIds.add(serverId);
	}
	public ComputerId getClientId() {
		return clientId;
	}
	public void setClientId(ComputerId clientId) {
		this.clientId = clientId;
	}
	public DistributedActivity<Object, Object> getDistributedActivity() {
		return distributedActivity;
	}
	public Activity<Object, Object> getActivity() {
		return distributedActivity.getActivity();
	}
	public void setDistributedActivity(DistributedActivity<Object, Object> activity) {
		this.distributedActivity = activity;
	}
	@Override
	public Channel<Object> getInput() {
		return inputChannel;
	}
	public void setInput(Channel<Object> inputChannel) {
		this.inputChannel = inputChannel;
	}
	@Override
	public Channel<Object> getOutput() {
		return outputChannel;
	}
	public void setOutput(Channel<Object> outputChannel) {
		this.outputChannel = outputChannel;
	}
	public void setAlive(boolean alive) {
		if(!alive) stopTime = System.currentTimeMillis();
		this.alive = alive;
	}
	@Override
	public long getLiveTime() {
		return alive? System.currentTimeMillis() - startTime:
			stopTime - startTime;
	}
	public void startRequest(){
		lastRequestStartTime = System.currentTimeMillis();
	}
	public void endRequest(){
		if(lastRequestStartTime!=0){
			final long now= System.currentTimeMillis();
			executionTime += now - lastRequestStartTime;
			lastRequestStartTime = now;
		}
		++callCount;
	}
	@Override
	public long getExecutionTime() {
		return executionTime;
	}

	@Override
	public long getCallCount() {
		return callCount;
	}
	@Override
	public boolean isAlive() {
		return alive;
	}
	@Override
	public void kill(Throwable error) {
		if(inputChannel == null && outputChannel == null){ 
			Log.error("Unhandled error", error);
		}
		if(inputChannel != null) inputChannel.fail(error);
		if(outputChannel != null) outputChannel.fail(error);
	}
	@Override
	public Object waitForResult(long timeoutMillis)
			throws InvocationTargetException, InterruptedException,
			TimeoutException {
		Lock<Object> lock = outputChannel.waitForReadLock(1, timeoutMillis);
		if(lock!=null){
			try{
				return lock.get(0);
			} finally{
				lock.unlock();
			}
		}
		if(outputChannel.hasFailed()){
			throw new InvocationTargetException(outputChannel.getFailureCause());
		}
		if(outputChannel.isClosed()){
			throw new IllegalStateException();
		}
		throw new TimeoutException(String.valueOf(timeoutMillis));
	}
	@Override
	public ActivityMonitor<?, ?> getParent() {
		// TODO Auto-generated method stub
		return null;
	}
	@Override
	public String toString() {
		return "RemoteActivityContext [activityId=" + activityId
				+ ", servers=" + serverIds + ", clientId=" + clientId
				+ ", activity=" + distributedActivity + ", executionTime=" + executionTime
				+ ", callCount=" + callCount + "]";
	}
	
	
}
